package dataconsume;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.net.URL;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.Properties;
import java.util.Scanner;

import org.apache.kafka.clients.producer.*;
import org.mozilla.universalchardet.UniversalDetector;

public class DataSender {

    private static final String[] AVAILABLE_FILES = {
            "题目资源.csv",
            "学生成绩预测.csv"
    };

    public static void main(String[] args) {
        String selectedFile = selectFile();
        if (selectedFile == null) return;

        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {
            URL resource = DataSender.class.getClassLoader().getResource(selectedFile);
            if (resource == null) {
                System.err.println("错误：文件 '" + selectedFile + "' 未找到");
                return;
            }

            try (InputStream inputStream = resource.openStream()) {
                // 自动检测文件编码
                String encoding = detectCharset(inputStream);
                System.out.println("检测到文件编码: " + encoding);

                // 使用正确编码重新打开流
                try (BufferedReader br = new BufferedReader(
                        new InputStreamReader(resource.openStream(), Charset.forName(encoding)))) {

                    System.out.println("\n正在发送文件: " + selectedFile);

                    // 跳过标题行
                    if (br.readLine() == null) {
                        System.err.println("错误：文件内容为空");
                        return;
                    }

                    String line;
                    int lineCount = 0;
                    while ((line = br.readLine()) != null) {
                        lineCount++;
                        ProducerRecord<String, String> record =
                                new ProducerRecord<>("javaTopic", line);

                        producer.send(record, (metadata, e) -> {
                            if (e != null) {
                                System.err.println("发送失败: " + e.getMessage());
                            }
                        });

                        System.out.printf("正在发送第 %d 条数据 [%s]%n",
                                lineCount,
                                abbreviate(line, 30));

                        Thread.sleep(100);
                    }
                    System.out.printf("\n✅ 成功发送 %d 条数据%n", lineCount);
                }
            } catch (InterruptedException e) {
                System.err.println("发送过程被中断");
                Thread.currentThread().interrupt();
            }
        } catch (IOException e) {
            System.err.println("文件读取错误: " + e.getMessage());
        }
    }

    // 自动检测文件编码
    private static String detectCharset(InputStream is) throws IOException {
        UniversalDetector detector = new UniversalDetector(null);
        byte[] buf = new byte[4096];

        int nread;
        while ((nread = is.read(buf)) > 0 && !detector.isDone()) {
            detector.handleData(buf, 0, nread);
        }
        detector.dataEnd();

        String encoding = detector.getDetectedCharset();
        detector.reset();

        // 默认使用UTF-8
        return encoding != null ? encoding : StandardCharsets.UTF_8.name();
    }

    // 简化长文本显示
    private static String abbreviate(String str, int maxWidth) {
        if (str.length() <= maxWidth) return str;
        return str.substring(0, maxWidth) + "...";
    }

    // 文件选择逻辑
    private static String selectFile() {
        Scanner scanner = new Scanner(System.in);
        System.out.println("请选择要发送的数据文件：");
        for (int i = 0; i < AVAILABLE_FILES.length; i++) {
            System.out.printf("[%d] %s%n", i + 1, AVAILABLE_FILES[i]);
        }
        System.out.print("请输入选项数字：");

        while (true) {
            try {
                int choice = Integer.parseInt(scanner.nextLine().trim());
                if (choice >= 1 && choice <= AVAILABLE_FILES.length) {
                    return AVAILABLE_FILES[choice - 1];
                }
                System.out.print("输入无效，请重新输入：");
            } catch (NumberFormatException e) {
                System.out.print("请输入有效数字：");
            }
        }
    }
}